package com.bsj.travel.forward.consumer.rocketmq;

import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.bsj.travel.cached.redis.RedisCached;
import com.bsj.travel.config.rocketmq.constant.ConsumerGroupConstant;
import com.bsj.travel.constant.BaseServerTopicConstants;
import com.bsj.travel.constant.MqttTopicPowerConstants;
import com.bsj.travel.constant.NumericalConstants;
import com.bsj.travel.constants.RedisConstant;
import com.bsj.travel.def.common.DO.SubscribeDO;
import com.bsj.travel.def.common.VO.device.DeviceInfoVO;
import com.bsj.travel.def.common.VO.device.HttpMessagePushVO;
import com.bsj.travel.def.global.mq.Rp;
import com.bsj.travel.forward.utils.OkHttpUtil;
import com.bsj.travel.util.AesUtils;
import com.bsj.travel.util.ExceptionUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @author LiJiaXin
 * @version 1.0.0
 * @description ProductSubscribeMessageRocketmqConsumer
 * @time 2024/2/29 16:51
 */
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = ConsumerGroupConstant.PRODUCT_SUBSCRIBE_GROUP,
        topic = BaseServerTopicConstants.PRODUCT_SUBSCRIBE_TOPIC,
        selectorExpression = BaseServerTopicConstants.DEVICE_STATE_CHANGE_TAG
                + "||" + BaseServerTopicConstants.DEVICE_LIFE_CYCLE_CHANGE_TAG
                + "||" + BaseServerTopicConstants.DEVICE_REPORT_INFO_TAG,
        messageModel = MessageModel.CLUSTERING)
public class ProductSubscribeMessageRocketmqConsumer implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {

    @Autowired
    private RedisCached redisCached;

    private final OkHttpUtil okHttpUtil = new OkHttpUtil();

    @Async("travelAsync")
    @Override
    public void onMessage(MessageExt ext) {
        if (ext == null) {
            log.info("产品订阅消息为空topic：{}", BaseServerTopicConstants.PRODUCT_SUBSCRIBE_TOPIC);
            return;
        }
        String message = new String(ext.getBody());
        log.info("产品订阅消息：{}", message);
        Rp<DeviceInfoVO> rp = JSON.parseObject(message, Rp.class);
        String tag = rp.getTag();
        DeviceInfoVO deviceInfoVO = JSONObject.parseObject(JSONObject.toJSONString(rp.getData()), DeviceInfoVO.class);
        String productKey = rp.getProductKey();
        if (StringUtils.isBlank(productKey)) {
            log.info("产品订阅————产品KEY为空");
            return;
        }
        try {
            String str = redisCached.get(RedisConstant.PRODUCT_SUBSCRIBE_INFO_KEY + productKey);
            if (StringUtils.isNotBlank(str)) {
                SubscribeDO subscribeDO = JSON.parseObject(str, SubscribeDO.class);
                if (subscribeDO == null) {
                    log.info("redis不为空但产品订阅为空：{}", str);
                    return;
                }
                List<Integer> pushTypes = Arrays.stream(subscribeDO.getPushType().split(",")).map(Integer::parseInt).collect(Collectors.toList());
                HttpMessagePushVO messagePushVO = new HttpMessagePushVO();
                messagePushVO.setProductKey(productKey);
                switch (tag) {
                    //设备状态变更tag
                    case BaseServerTopicConstants.DEVICE_STATE_CHANGE_TAG:
                        if (pushTypes.contains(NumericalConstants.ONE)) {
                            messagePushVO.setMessageType(NumericalConstants.ONE);
                        }
                        break;
                    //设备生命周期变更tag
                    case BaseServerTopicConstants.DEVICE_LIFE_CYCLE_CHANGE_TAG:
                        if (pushTypes.contains(NumericalConstants.TWO)) {
                            messagePushVO.setMessageType(NumericalConstants.TWO);
                        }
                        break;
                    //设备上报信息tag
                    case BaseServerTopicConstants.DEVICE_REPORT_INFO_TAG:
                        if (pushTypes.contains(NumericalConstants.THREE)) {
                            messagePushVO.setMessageType(NumericalConstants.THREE);
                        }
                        break;
                    default:
                        log.info("未知产品订阅tag: {}", tag);
                        return;
                }
                messagePushVO.setDeviceInfoVO(deviceInfoVO);
                String url = subscribeDO.getUrl();
                String jsonString = JSON.toJSONString(messagePushVO);
                String secretKey = subscribeDO.getSecretKey();
                if (!subscribeDO.getIsEncrypt().equals(NumericalConstants.ONE)) {
                    jsonString = AesUtils.encrypt(secretKey, jsonString);
                }
                okHttpUtil.doPost(url, null, Map.of("message", jsonString));
            }
        } catch (Exception e) {
            log.error("产品订阅数据-->消费失败，失败原因: {}", ExceptionUtil.getStackStr(e));
        }
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        // 设置消费者消息重试次数
        defaultMQPushConsumer.setMaxReconsumeTimes(3);
        //设置实例名称
        defaultMQPushConsumer.setInstanceName(MqttTopicPowerConstants.PRODUCT_SUBSCRIBE_INSTANCE_NAME);
    }
}